package flow;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;

import java.io.IOException;

public class MyRecordReader extends RecordReader<Flow, NullWritable> {
    LineRecordReader lineRecordReader = new LineRecordReader();
    Flow flow = new Flow();
    String split = " ";
    @Override
    public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        lineRecordReader.initialize(inputSplit,taskAttemptContext);
    }

    /**
     * 决定 key和value的值
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        Text currentValue = null;
        if (lineRecordReader.nextKeyValue()){
            currentValue = lineRecordReader.getCurrentValue();
            String[] splits = currentValue.toString().split(this.split);
            System.out.println(splits.toString());
            System.out.println(splits[0]+" "+splits[1]+" "+splits[2]+" "+splits[3]);
            flow.setPhone(splits[0]);
            flow.setUpFlow(Integer.parseInt(splits[1].toString()));
            flow.setDownFlow(Integer.parseInt(splits[2].toString()));
            flow.setSumFlow(Integer.parseInt(splits[3].toString()));
        }
        if (currentValue == null){
            return false;
        }
        return true;
    }

    @Override
    public Flow getCurrentKey() throws IOException, InterruptedException {
        return flow;
    }

    @Override
    public NullWritable getCurrentValue() throws IOException, InterruptedException {
        return null;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {

    }
}
